/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 */

package org.apache.skywalking.oap.server.core.analysis.metrics;

import java.util.*;
import lombok.*;
import org.apache.skywalking.oap.server.core.analysis.metrics.annotation.*;
import org.apache.skywalking.oap.server.core.query.sql.Function;
import org.apache.skywalking.oap.server.core.storage.annotation.Column;

/**
 * PxxMetrics is a parent metrics for p99/p95/p90/p75/p50 metrics. P(xx) metrics is also for P(xx) percentile.
 *
 * A percentile (or a centile) is a measure used in statistics indicating the value below which a given percentage of
 * observations in a group of observations fall. For example, the 20th percentile is the value (or score) below which
 * 20% of the observations may be found.
 *
 * @author wusheng, peng-yongsheng
 */
// 计算的是时间窗口内的分位数，例如： P99Metrics、P95Metrics、P70Metrics等
// Pxx 这个指标的含义，这里以 P90 为例说明：计算 P90 是在一个监控序列中，找到一个最小的值，大于序列中 90% 的监控值。

// 根据对 Model 的介绍，每个 AllP*Metrics 在 ES 中有四个对应的 Index 名称分别是：
// all_p*-20200115
// all_p*_hour-20200115
// all_p*_day-20200115
// all_p*_month-202001

//  AllP* Metrics 在 ElasticSearch 存储的 Document 内容：
// {
//   "_index": "all_p90-20191209",  # Index 名称
//   "_type": "type",
//   "_id": "201912091056", # Document Id
//   "_version": 2,
//   "_score": 1,
//   "_source": {
//     "precision": 10, # 精度，10ms 精度
//     "time_bucket": 201912091056, # 时间窗口
//     "value": 2010,   # 计算之后的value值，即 P90 值
//     "detail_group": "200,1|201,1" # 该时间窗口内的全部监控数据(10ms精度)
//   }
// }
public abstract class PxxMetrics extends Metrics implements IntValueHolder {
    protected static final String DETAIL_GROUP = "detail_group";
    protected static final String VALUE = "value";
    protected static final String PRECISION = "precision";

    // 计算之后的监控结果
    @Getter @Setter @Column(columnName = VALUE, isValue = true, function = Function.Avg) private int value;
    // 监控值的精度，默认是10毫秒级的监控
    @Getter @Setter @Column(columnName = PRECISION) private int precision;
    // 记录当前监控在时间窗口内的全部数据，IntKeyLongValueArray 继承了 ArrayList，其中每个元素都是 IntKeyLongValue 类型
    @Getter @Setter @Column(columnName = DETAIL_GROUP) private IntKeyLongValueArray detailGroup;

    // 分位数，例如，P99Metrics 中该字段值为 99，P90Metrics 中该字段值为90
    private final int percentileRank;
    // 用于合并相同监控值，其中 key 为监控值(降精度的)，value 记录了对应的IntKeyLongValue对象
    private Map<Integer, IntKeyLongValue> detailIndex;

    public PxxMetrics(int percentileRank) {
        this.percentileRank = percentileRank;
        detailGroup = new IntKeyLongValueArray(30);
    }

    @Entrance
    public final void combine(@SourceFrom int value, @Arg int precision) {
        this.precision = precision; // 确定监控精度，默认为10，即10毫秒级别

        this.indexCheckAndInit(); // 初始化 detailIndex这个Map

        int index = value / precision;
        IntKeyLongValue element = detailIndex.get(index);
        if (element == null) { // 创建 IntKeyLongValue 对象
            element = new IntKeyLongValue();
            element.setKey(index);
            element.setValue(1);
            // 记录到 detailGroup 和 detailIndex 集合中
            addElement(element);
        } else {
            element.addValue(1); // 递增 value
        }
    }

    @Override
    public void combine(Metrics metrics) {
        PxxMetrics pxxMetrics = (PxxMetrics)metrics;
        this.indexCheckAndInit();
        pxxMetrics.indexCheckAndInit();

        pxxMetrics.detailIndex.forEach((key, element) -> {
            IntKeyLongValue existingElement = this.detailIndex.get(key);
            if (existingElement == null) {
                existingElement = new IntKeyLongValue();
                existingElement.setKey(key);
                existingElement.setValue(element.getValue());
                addElement(element);
            } else {
                existingElement.addValue(element.getValue());
            }
        });
    }

    @Override
    public final void calculate() {
        // 排序detailGroup
        Collections.sort(detailGroup);
        // 计算该窗口监控点的总个数
        int total = detailGroup.stream().mapToInt(element -> (int)element.getValue()).sum();
        // 查找指定分位数的位置
        int roof = Math.round(total * percentileRank * 1.0f / 100); // round 方法：返回与实参最接近的整数

        int count = 0;
        for (IntKeyLongValue element : detailGroup) {
            // 累加监控点个数，直至到达(或超过)上面的 roof 值，此时的监控值即为指定分位数监控值
            count += element.getValue();
            if (count >= roof) {
                value = element.getKey() * precision;
                return;
            }
        }
    }

    private void addElement(IntKeyLongValue element) {
        detailGroup.add(element);
        detailIndex.put(element.getKey(), element);
    }

    private void indexCheckAndInit() {
        if (detailIndex == null) {
            detailIndex = new HashMap<>();
            detailGroup.forEach(element -> detailIndex.put(element.getKey(), element));
        }
    }
}
